/*******************************************************************************
 * Copyright 2010 DemandSpot
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 ******************************************************************************/
package com.demandspot.aws;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.annotation.PostConstruct;

import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.springframework.beans.factory.annotation.Autowired;

import com.xerox.amazonws.sdb.Domain;
import com.xerox.amazonws.sdb.Item;
import com.xerox.amazonws.sdb.ItemAttribute;
import com.xerox.amazonws.sdb.SDBException;
import org.apache.log4j.Logger;

public class SDBNodeDiscoveryService {
	private static final Logger logger = Logger.getLogger(SDBNodeDiscoveryService.class);

	public static enum AttributeName {nodeName, hostName, address, heartbeat, period}

	@Autowired private AWSSDBService sdbService;
	
	private DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyyMMddHHmmss").withZone(DateTimeZone.UTC);
	private String domainName;
	private Domain domain;
	private SDBNodeData data = new SDBNodeData();
	private boolean master;

	public SDBNodeDiscoveryService(String nodeName, boolean master, int devMode, long period, String domainName) throws UnknownHostException {
		this(nodeName, master, devMode == 1, period, domainName);
	}
	
	public SDBNodeDiscoveryService(String nodeName, boolean master, String devMode, long period, String domainName) throws UnknownHostException {
		this(nodeName, master, "1".equals(devMode), period, domainName);
	}
	
	public SDBNodeDiscoveryService(String nodeName, boolean master, boolean devMode, long period, String domainName) throws UnknownHostException {
		this.data.setNodeName(nodeName);
		this.data.setHostName(InetAddress.getLocalHost().getHostName());
		this.data.setAddress(InetAddress.getLocalHost().getHostAddress());
		this.data.setKey(this.data.getNodeName() + "_" + this.data.getHostName());
		this.data.setPeriod(period);
		this.master = master;
		this.domainName = domainName;
		if (devMode)
			this.domainName += "_" + System.getProperty("user.name");
	}
	
	@PostConstruct
	void init() throws SDBException {
		domain = sdbService.getOrCreateDomain(domainName);
		beat();
	}

	private void updateItem() throws SDBException {
		List<ItemAttribute> attrs = createItemAttributesList();
		
		if (data.item == null) {
			Item item = domain.getItem(data.getKey());
			item.putAttributes(attrs);
			data.item = item;
		} else {
			data.item.putAttributes(attrs);
		}
	}

	private List<ItemAttribute> createItemAttributesList() {
		DateTime now = new DateTime();
		
		List<ItemAttribute> attrs = new ArrayList<ItemAttribute>();
		attrs.add(new ItemAttribute(AttributeName.nodeName.name(), data.getNodeName(), true));
		attrs.add(new ItemAttribute(AttributeName.hostName.name(), data.getHostName(), true));
		attrs.add(new ItemAttribute(AttributeName.heartbeat.name(), formatter.print(now), true));
		attrs.add(new ItemAttribute(AttributeName.address.name(), data.getAddress(), true));
		attrs.add(new ItemAttribute(AttributeName.period.name(), String.valueOf(data.getPeriod()), true));
		
		this.data.setLastHeartbeat(now);
		
		return attrs;
	}
	
	public void beat() throws SDBException {
		updateItem();
		logger.debug("Sent heartbeat to SimpleDB:" + domain.getName());
		
		if (master) {
			deleteOldEntries();
		}
	}
	
	public void deleteOldEntries() throws SDBException {
		Map<String, List<ItemAttribute>> items = sdbService.getAllItemsWithAttributes(domain);
		Map<String, List<SDBNodeData>> map = new HashMap<String, List<SDBNodeData>>();
		
		for (List<ItemAttribute> attrs : items.values()) {
			SDBNodeData nodeData = createNodeData(attrs);
			List<SDBNodeData> nodes = map.get(nodeData.getNodeName());
			if (nodes == null) {
				nodes = new ArrayList<SDBNodeData>();
				map.put(nodeData.getNodeName(), nodes);
			}
			nodes.add(nodeData);
		}
		
		for (Entry<String, List<SDBNodeData>> entry : map.entrySet()) {
			List<SDBNodeData> list = entry.getValue();
			
			if (list.size() > 1) {
				Collections.sort(list, new Comparator<SDBNodeData>() {
					@Override
					public int compare(SDBNodeData o1, SDBNodeData o2) {
						return o1.compareTo(o2) * -1;
					}
				});
				for(int i=0; i < list.size() - 1; i++) {
					SDBNodeData nodeData = list.get(i);
					nodeData.setKey(nodeData.nodeName + "_" + nodeData.hostName);
					long last = nodeData.lastHeartbeat.getMillis();
					long period = nodeData.period * 2;
					boolean delete = System.currentTimeMillis() - last > period;
					
					if (delete) {
						domain.deleteItem(nodeData.key);
					} else {
						break;
					}
				}
			}
		}
	}
	
	private SDBNodeData createNodeData(List<ItemAttribute> attrs) throws SDBException{
		SDBNodeData nodeData = new SDBNodeData();
		for (ItemAttribute itemAttribute : attrs) {
			String name = itemAttribute.getName();
			String value = itemAttribute.getValue();
			
			AttributeName attributeName = AttributeName.valueOf(name);
			switch (attributeName) {
			case hostName:
				nodeData.setHostName(value);
				break;
			case heartbeat:
				nodeData.setLastHeartbeat(formatter.parseDateTime(value));
				break;
			case nodeName:
				nodeData.setNodeName(value);
				break;
			case address:
				nodeData.setAddress(value);
				break;
			case period:
				nodeData.setPeriod(Long.parseLong(value));
				break;
			}
		}
		return nodeData;
	}

	/*
	 * Retrieve all current nodes on SimpleDB
	 */
	public List<SDBNodeData> listNodes() throws SDBException {
		Map<String, List<ItemAttribute>> items = sdbService.getAllItemsWithAttributes(domain);
		List<SDBNodeData> list = new ArrayList<SDBNodeData>();
		
		for (List<ItemAttribute> attrs : items.values())
			list.add(createNodeData(attrs));
		
		Collections.sort(list, new Comparator<SDBNodeData>() {

			@Override
			public int compare(SDBNodeData o1, SDBNodeData o2) {
				int res = o1.nodeName.compareTo(o2.nodeName);
				
				if (res == 0 && o1.nodeName.equals(o2.nodeName))
					res = o2.lastHeartbeat.compareTo(o1.lastHeartbeat);
				
				return res;
			}
		});
		
		return list;
	}
	
	public static class SDBNodeData implements Comparable<SDBNodeData>{
		private String nodeName;
		private String hostName;
		private String address;
		private long period = -1;
		private DateTime lastHeartbeat;
		private String key;
		private Item item;
		
		public SDBNodeData() {
		}
		
		public String getNodeName() {
			return nodeName;
		}
		
		public void setNodeName(String nodeName) {
			this.nodeName = nodeName;
		}
		
		public String getHostName() {
			return hostName;
		}
		
		public void setHostName(String address) {
			this.hostName = address;
		}
		
		public DateTime getLastHeartbeat() {
			return lastHeartbeat;
		}
		
		public void setLastHeartbeat(DateTime lastHeartbeat) {
			this.lastHeartbeat = lastHeartbeat;
		}
		
		public String getKey() {
			return key;
		}
		
		public void setKey(String key) {
			this.key = key;
		}

		public Item getItem() {
			return item;
		}

		public void setItem(Item item) {
			this.item = item;
		}

		public String getAddress() {
			return address;
		}

		public void setAddress(String address) {
			this.address = address;
		}

		public long getPeriod() {
			return period;
		}

		public void setPeriod(long period) {
			this.period = period;
		}

		@Override
		public int compareTo(SDBNodeData o) {
			return o.lastHeartbeat.compareTo(lastHeartbeat);
		}
	}
}
